package com.offcn.spark.p3

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Auther: BigData-LGW
 * @ClassName: BroadcastVariable
 * @Date: 2020/12/8 18:32
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
object BroadcastVariable {
//    def main(args: Array[String]): Unit = {
//        val conf = new SparkConf()
//            .setAppName("_01BroadcastVariableOps")
//            .setMaster("local[*]")
//        val sc = new SparkContext(conf)
//        val genderMap = Map(
//            "0" -> "妹砸儿",
//            "1" -> "大兄弟"
//        )
//
//        val stuRDD = sc.parallelize(List(
//            Student("01", "宋敏健", "0", 18),
//            Student("02", "严文青", "1", 19),
//            Student("03", "王大伟", "1", 18),
//            Student("04", "闫来宾", "1", 22)
//        ))
//        stuRDD.map(stu => {
//            val gender = stu.gender
//            Student(stu.id,stu.name,genderMap.getOrElse(gender,"ladyBoy"),stu.age)
//        }).foreach(println)
//        val genderBC:Broadcast[Map[String,String]] = sc.broadcast(genderMap)
//        stuRDD.map(stu => {
//            val gender = genderBC.value.getOrElse(stu.gender,"ladyBoy")
//            Student(stu.id,stu.name,gender,stu.age)
//        }).foreach(println)
//        sc.stop()
//    }
//}
//case class Student(id: String, name: String, gender: String, age: Int)
def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
        .setAppName("_01BroadcastVariableOps")
        .setMaster("local[*]")
    val sc = new SparkContext(conf)

    val stus = List(
        Student(1, "唐玉峰", "安徽·合肥"),
        Student(2, "李梦", "山东·济宁"),
        Student(3, "胡国权", "甘肃·白银"),
        Student(4, "陈延年", "甘肃·张掖"),
        Student(5, "马惠", "辽宁·葫芦岛"),
        Student(10086, "刘炳文", "吉林·通化")
    )
    val scoreRDD = sc.parallelize(List(
        Score(1, "chinese", 95.5),
        Score(2, "english", 55.5),
        Score(3, "math", 20.5),
        Score(4, "pe", 32.5),
        Score(5, "physical", 59),
        Score(10000, "Chemistry", 99.5)
    ))

    //        joinOps(stus, scoreRDD)
    broadcastOps(stus, scoreRDD)
    sc.stop()
}
    /*
        使用广播变量的方式来完成如下的关联操作
        map join-->大表+小表
     */
    def broadcastOps(stus: List[Student], scoreRDD:RDD[Score]): Unit = {
        val id2Stu = stus.map(stu => (stu.id, stu)).toMap
        //构建广播变量
        val bc: Broadcast[Map[Int, Student]] = scoreRDD.sparkContext.broadcast(id2Stu)

        scoreRDD.foreach(score => {
            val id = score.sid
            val stu = bc.value.getOrElse(id, Student(-1, null, null))
            println(s"${stu.id}\t${stu.name}\t${stu.province}\t${score.course}\t${score.score}")
        })
    }

    def joinOps(stus: List[Student], scoreRDD:RDD[Score]): Unit = {
        val stuRDD = scoreRDD.sparkContext.parallelize(stus)
        val id2Stu:RDD[(Int, Student)] = stuRDD.map(stu => (stu.id, stu))
        val id2Score:RDD[(Int, Score)] = scoreRDD.map(score => (score.sid, score))
        val joinedRDD:RDD[(Int, (Student, Score))] = id2Stu.join(id2Score)
        joinedRDD.foreach{case (id, (stu, score)) => {
            println(s"id为${id}的学生信息为:${stu}，其考试成绩信息为：${score}")
        }}
    }
}
case class Student(id: Int, name:String, province: String)
case class Score(sid: Int, course: String, score: Double)